package communication.common.business.heartbeat;


import communication.common.business.base.*;
import communication.common.business.util.MsgDataUtil;
import communication.common.vertx.net.ConnectRecord;
import communication.common.vertx.net.NetManager;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;
import org.springframework.util.StringUtils;

import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.*;

/**
 * client端的心跳处理服务，主要是用来 :发送心跳, 处理心跳应答 ，心跳发送间隔检查
 * @author mac
 */
@Slf4j
@Data
public abstract class HeartBeatServiceCommon {

    private NetManager netManager;

    private FcrAutoConfig fcrAutoConfig;


    /**
     * 心跳记录。
     */
    private  Map<FcrSocketAddress,RecordHeartBeat> heartBeatMap = new ConcurrentHashMap<>();

    /**
     * 超时线程存活时间，毫秒
     */
    private long keepAliveTime=5000;

    /**
     * 线程池用的队列大小
     */
    private int queueSize = 20;

    /**
     * 执行定时任务的线程池
     */
    private ThreadPoolExecutor threadPoolExecutor ;







    /**
     * 有参构造
     * @param netManager 所有socket的连接管理
     */
    public HeartBeatServiceCommon(NetManager netManager,FcrAutoConfig fcrAutoConfig) throws Exception {
        this.netManager = netManager;
        this.fcrAutoConfig = fcrAutoConfig;
        int processors = Runtime.getRuntime().availableProcessors();
        BlockingQueue<Runnable> workQueue = new ArrayBlockingQueue<>(queueSize);
        ThreadFactory springThreadFactory = new CustomizableThreadFactory("HeartBeat-threadPool-");
        threadPoolExecutor = new ThreadPoolExecutor(processors, 2* processors, keepAliveTime, TimeUnit.MILLISECONDS , workQueue,springThreadFactory);

        // 执行定时任务
        setTask();
    }


    /**
     *
     *  启动心跳检查
     *  心跳周期发送检查
     *  默认每隔10秒发送一次
     *
     */
    public void  checkPeriodAllHeartBeat() {
        // 工作模式为client才执行
        if( !fcrAutoConfig.getWorkMode().equals(WorkModeEnum.CLIENT)){
            return;
        }

        HeartBeatConfig heartBeatConfig = fcrAutoConfig.getHeartBeat();
        long onTime = System.currentTimeMillis()+heartBeatConfig.getInterval();

        // 停止标识为false才执行task
        while (true){
            long now = System.currentTimeMillis();
            if( now >= onTime ){
                onTime = System.currentTimeMillis()+heartBeatConfig.getInterval();
                try {
                    // 根据协议类型，获取对应协议的socket连接的记录
                    Iterator iterator = null;

                    switch (fcrAutoConfig.getProtocol()){
                        case TCP:
                            iterator = this.getNetManager().getTcpSocketIterator();
                            break;
                        case UDP:
                            iterator = this.getNetManager().getUdpSocketIterator();
                            break;
                        default:
                            // nothing
                    }

                    if( iterator == null){
                        return;
                    }
                    checkSocketItAndSendHeartBeat( iterator );
                } catch (Exception e) {
                    e.printStackTrace();
                    log.warn("[发生异常, endSchedule flag set true !]");
                }
            }
        }
    }




    /**
     * 检查socket的迭代器，如果检查通过则发送心跳
     *
     * @param iterator tcp类型的socket的迭代器
     */
    private void checkSocketItAndSendHeartBeat(Iterator<Map.Entry<FcrSocketAddress,Object>> iterator) throws Exception {
        while (iterator.hasNext()){

            Map.Entry<FcrSocketAddress, Object> next = iterator.next();

            ConnectRecord connectRecord = (ConnectRecord)next.getValue();

            RecordHeartBeat record = heartBeatMap.get(next.getKey());

            // 检查是否需要发送心跳
            Integer seq = checkNeedSendHeartAndGetSeq(record , connectRecord.getRemoteMode());

            // 不发送心跳，则去除记录的socket连接与心跳记录
            if(seq == null){
                // 清除记录的超时次数
                heartBeatMap.remove( next.getKey());
                iterator.remove();
                continue;
            }

            // socket存在，进行心跳处理
            sendHeartBeat(next.getKey(),seq);
        }

    }


    /**
     * 发送心跳
     * 想指定的主机、端口发送心跳包，然后记录心跳数据
     */
    public void sendHeartBeat( FcrSocketAddress remoteSocketAddress, int seq) throws Exception {

        // 获取当前时间
        long now =  System.currentTimeMillis();

        // 构建心跳数据
        HeartBeat heartBeat = HeartBeat.builder()
                .seq(seq)
                .interval(fcrAutoConfig.getHeartBeat().getInterval())
                .sendTime(now)
                .build();

        MsgData msgData = MsgDataUtil.createHeartBeat(heartBeat);

        // 发送之前就记录，避免还未记录心跳数据时就收到了心跳的ACK
        RecordHeartBeat record = recordHeartBeat(now, remoteSocketAddress.getHost(), remoteSocketAddress.getPort(), heartBeat);

        switch (record.getProtocolEnum()){
            case TCP:
                // send tcp
                this.netManager.sendMsgDataByTcpClient(msgData,remoteSocketAddress);
                break;
            case UDP:
                // send udp
                this.netManager.sendMsgDataByUdpServer(msgData,remoteSocketAddress,null);
                break;
            default:
                // nothing
        }

    }



    /**
     * 检查是否需要发送心跳包;
     *  1. 如果是第一次发送心跳包，则返回序号0；
     *  2. 超时次数 <  配置的最大超时次数, 则返回 序号+1
     *  3. 超时次数 >=  配置的最大超时次数,则返回Null
     *
     * @param record 心跳记录
     * @param remoteMode
     * @return null 或者 心跳包序号
     */
    private Integer checkNeedSendHeartAndGetSeq(RecordHeartBeat record, WorkModeEnum remoteMode){

        //  对端的连接模式为null 或者 不是server模式，则不发送心跳
        if( remoteMode == null ||  ! remoteMode.equals(WorkModeEnum.SERVER)){
            return null;
        }


        // 没有发送记录，可以发送心跳;
        if(record==null || record.getHeartBeat()==null) {
            return 0;
        }

        // 超时次数 <  配置的最大超时次数
        if(record.getTimeOutCount() < this.fcrAutoConfig.getHeartBeat().getMaxTimeOut()){
            return record.getHeartBeat().getSeq()+1;
        }else{
            // 超时次数 > 限制, 不发送心跳。去除记录的socket连接
            log.warn("心跳超时次数超过限制! 当前超时次数: "+record.getTimeOutCount());
            return null;
        }

    }


    /**
     * 记录心跳发送记录, 默认超时次数+1,收到响应时，再清空超时次数
     * @param host 心跳发往的主机
     * @param port 心跳发往的端口
     * @param time 心跳发送的时间
     * @param heartBeat 心跳数据
     */
    public RecordHeartBeat recordHeartBeat(Long time ,String host,Integer port,HeartBeat heartBeat) throws Exception {

        // 参数检查
        if(time==null){
            throw new Exception(" record heartBeat fail , time is null !");
        }

        if(StringUtils.isEmpty(host)){
            throw new Exception(" record heartBeat fail , host is null !");
        }

        if(port==null){
            throw new Exception(" record heartBeat fail , port is null !");
        }

        FcrSocketAddress remoteSocketAddress = new FcrSocketAddress(host,port);

        RecordHeartBeat record = heartBeatMap.get(remoteSocketAddress);

        if(record==null){
            record= RecordHeartBeat.builder()
                    .host(host)
                    .port(port)
                    .protocolEnum(this.fcrAutoConfig.getProtocol())
                    .build();

        }
        // 每一次发送，超时次数都+1，只有收到ack时才回清零超时次数
        record.setTimeOutCount(record.getTimeOutCount()+1);
        record.setHeartBeat(heartBeat);

        heartBeatMap.put(remoteSocketAddress,record);

        return record;
    }


    /**
     * 处理心跳应答
     */
    public void receiveAck(HeartBeatAck ack, FcrSocketAddress remoteSocketAddress){

        RecordHeartBeat record = heartBeatMap.get(remoteSocketAddress);

        // 先收到应答，心跳还未记录
        if(record==null || record.getHeartBeat()==null){
            log.warn("heartBeat record is null,can not record ack!");
            return;
        }

        log.debug(" received heartBeatAck :{}",ack);

        // ack.seq < record.seq，说明网络不稳定，可能出现ack掉包 或者 大序号的ack先发送到的情况
        if(ack.getAckSeq() < record.getHeartBeat().getSeq()){
            log.warn(" NetWork is unstable !!!  HeartBeatAck.seq ={} , HeartBeat.seq ={} ",ack.getAckSeq(),record.getHeartBeat().getSeq());
            return;
        }

        // 应答序号 > 心跳序号，则说明客户端处理错误或者发送错误，忽略
        if(ack.getAckSeq() > record.getHeartBeat().getSeq()){
            log.warn("  HeartBeatAck.seq ={}  more than  HeartBeat.seq ={} ",ack.getAckSeq(),record.getHeartBeat().getSeq());
            return;
        }

        // 应答序号 == 心跳序号，更新reocrd
        if(ack.getAckSeq() == record.getHeartBeat().getSeq()){
            record.setTimeOutCount(0);
            record.setHeartBeatAck(ack);
        }

    }


    /**
     * 向指定的 host:port 发送心跳响应
     * @param remoteSocketAddress 远端地址
     * @param ackSeq 心跳seq
     */
    private void sendAck(FcrSocketAddress remoteSocketAddress,int ackSeq) throws Exception {
        // 参数校验
        if(remoteSocketAddress == null){
            throw new Exception("send heartAck fail , remoteSocketAddress is null !");
        }

        // build ack
        HeartBeatAck ack = HeartBeatAck.builder()
                .ackSeq(ackSeq)
                .time(System.currentTimeMillis())
                .build();

        MsgData heartBeatAck = MsgDataUtil.createHeartBeatAck(ack);

        switch (this.fcrAutoConfig.getProtocol()){
            case TCP:
                this.netManager.sendMsgDataByTcpClient(heartBeatAck,remoteSocketAddress);
                break;
            case UDP:
                // send
                this.netManager.sendMsgDataByUdpServer(heartBeatAck,remoteSocketAddress,null);
                break;
            default:
        }

        RecordHeartBeat record = heartBeatMap.get(remoteSocketAddress);
        record.setHeartBeatAck(ack);
        heartBeatMap.put(remoteSocketAddress,record);
    }

    /**
     * 处理心跳. 存储心跳记录，发送心跳ack
     */
    public void handleHeartBeat(HeartBeat heartBeat , FcrSocketAddress remoteSocketAddress) throws Exception {

        log.debug("received heartBeat:{} , remoteSocketAddress:{}",heartBeat,remoteSocketAddress.getAddress());

        RecordHeartBeat record=null;

        if( heartBeatMap.containsKey(remoteSocketAddress)
                && heartBeatMap.get(remoteSocketAddress) != null ){
            record = heartBeatMap.get(remoteSocketAddress);
        }else{
            record = RecordHeartBeat.builder().build();
        }

        // save
        heartBeat.setReceivedTime(System.currentTimeMillis());
        record.setHeartBeat(heartBeat);
        record.setHost(remoteSocketAddress.getHost());
        record.setPort(remoteSocketAddress.getPort());

        // clear timeOutCount
        record.setTimeOutCount(0);
        heartBeatMap.put(remoteSocketAddress,record);

        // send ack
        sendAck(remoteSocketAddress,heartBeat.getSeq());
    }







    /**
     * 设置定时任务
     * @throws Exception
     */
    public abstract void setTask() throws Exception;

}
